package com.chinaums.format.deserializer;

import com.chinaums.format.option.SftpCsvOption;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.table.connector.RuntimeConverter.Context;
import org.apache.flink.table.connector.source.DynamicTableSource.DataStructureConverter;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.*;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.List;


public class SftpCsvDeserializer implements DeserializationSchema<RowData> {
    private static final Logger logger = LoggerFactory.getLogger(SftpCsvDeserializer.class);
    private static final String AND = "&&";
    private static final RowKind KIND = RowKind.valueOf("INSERT");
    private final List<LogicalType> parsingTypes;
    private final DataStructureConverter converter;
    private final TypeInformation<RowData> producedTypeInfo;

    private final boolean ignoreParseErrors;
    private final String columnDelimiter;
    private final String lineDelimiter;
    private String cDelimiter = "";
    private String lDelimiter = "";

    // 行分隔符缓存数据
    private final StringBuffer buffer = new StringBuffer();

    // true 为此行异常，需要与下一行连接起来处理
    private volatile boolean adjacent = false;


    // 移除字段前后的 \t
    private volatile String tab;

    public SftpCsvDeserializer(List<LogicalType> parsingTypes, DataStructureConverter converter, TypeInformation<RowData> producedTypeInfo, SftpCsvOption option) {
        this.parsingTypes = parsingTypes;
        this.converter = converter;
        this.producedTypeInfo = producedTypeInfo;
        this.columnDelimiter = option.getColumnDelimiter();
        this.ignoreParseErrors = option.isIgnoreParseErrors();
        this.lineDelimiter = option.getLineDelimiter();
    }


    @Override
    public TypeInformation<RowData> getProducedType() {
        return producedTypeInfo;
    }

    @Override
    public void open(InitializationContext context) {
        converter.open(Context.create(SftpCsvDeserializer.class.getClassLoader()));
    }

    @Override
    public RowData deserialize(byte[] message) {
        if (message == null) {
            return null;
        }
        initParameter();

        String line = new String(message, StandardCharsets.UTF_8);

        // 获取:   ————>  不是以指定换行符结尾的数据 并记录下   // 003&&007
        if (StringUtils.isNotEmpty(lDelimiter) && !line.endsWith(lDelimiter)) {
            adjacent = true;
            buffer.append(line);
            return null;
        }
        try {
            return process(line);

        }catch (ArrayIndexOutOfBoundsException e){
            logger.error("it may be that the number of fields does not match the number of data (数组下标越界可能是字段与数据个数不匹配，请检查sql与数据个数是否一致)");
            throw new RuntimeException(e);
        } catch (Throwable t) {
            if (adjacent) {
                String value = buffer.append(line).toString();
                buffer.setLength(0);
                adjacent = false;
                return process(value);
            }
            if (ignoreParseErrors) {
                logger.error(">>> error data: [{}]", line);
                return null;
            }
            throw new RuntimeException(t);
        }


    }



    private RowData process(String line) {

        final Row row = new Row(KIND, parsingTypes.size());
        line = line + " ";
        final String[] columns = line.split(cDelimiter);

        for (int i = 0; i < parsingTypes.size(); i++) {

            String column = columns[i].replaceAll(tab, "");

            if (i == columns.length - 1 && StringUtils.isNotEmpty(lDelimiter)) {
                String replaceCol = column.replaceAll(lDelimiter, "");
                row.setField(i, parse(parsingTypes.get(i).getTypeRoot(), replaceCol.trim()));
            } else {
                row.setField(i, parse(parsingTypes.get(i).getTypeRoot(), column.trim()));
            }
        }
        return (RowData) converter.toInternal(row);

    }

    /**
     * 初始化参数
     */
    private void initParameter() {

        // 列分割符号 \003     003&&007   003 007  003&&007&&004
        if (StringUtils.isNotEmpty(columnDelimiter) && StringUtils.isBlank(cDelimiter)) {
            if (columnDelimiter.trim().startsWith("00")) {
                String[] columnSplit = columnDelimiter.split(AND);
                byte[] columnBytes = new byte[columnSplit.length];
                for (int i = 0; i < columnSplit.length; i++) {
                    columnBytes[i] = Byte.parseByte(columnSplit[i].substring(columnSplit[i].length() - 1));
                }
                cDelimiter = new String(columnBytes);
            }else {
                cDelimiter = columnDelimiter;
            }

        }

        // 行分割符号
        if (StringUtils.isNotEmpty(lineDelimiter) && StringUtils.isBlank(lDelimiter)) {
            String[] lineSplit = lineDelimiter.split(AND);
            byte[] lineBytes = new byte[lineSplit.length];
            for (int i = 0; i < lineSplit.length; i++) {
                lineBytes[i] = Byte.parseByte(lineSplit[i].substring(lineSplit[i].length() - 1));
            }
            lDelimiter = new String(lineBytes);
        }

        // 字段前后的制表符
        if (StringUtils.isBlank(tab)) {
            tab = new String(new byte[]{9});
        }


    }

    private static Object parse(LogicalTypeRoot root, String value) {
        switch (root) {
            case INTEGER:
                return Integer.parseInt(value);
            case VARCHAR:
                return value;
            default:
                throw new IllegalArgumentException();
        }

    }

    @Override
    public boolean isEndOfStream(RowData nextElement) {
        return false;
    }

}
